package cn.e4j.core.kettle;

import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.util.EnvUtil;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;

import java.util.Map;

/**
 * kettle任务调度类
 * Created by yuan_rz on 2019-03-12.
 */
public class KettleUtil {
    /**
     * 调用trans文件
     * @param transFileName
     * @throws Exception
     */
    public static void callNativeTrans(String transFileName) throws Exception{
        callNativeTransWithParams(null, transFileName);
    }

    /**
     * 调用trans文件 带参数的
     * @param params
     * @param transFileName
     * @throws Exception
     */
    public static void callNativeTransWithParams(Map<String,String> params , String transFileName) throws Exception{
        // 初始化
        KettleEnvironment.init();
        EnvUtil.environmentInit();
        TransMeta transMeta = new TransMeta(transFileName);
        //转换
        Trans trans = new Trans(transMeta);
        //参数
        if(params != null){
            for(Map.Entry<String,String> entry:  params.entrySet() ) {
                trans.setParameterValue(entry.getKey(), entry.getValue());
            }
        }
        //执行
        trans.execute(null);
        //等待结束
        trans.waitUntilFinished();
        //抛出异常
        if(trans.getErrors() > 0){
            throw new RuntimeException("Kettle执行转换发生异常");
        }
    }

    /**
     * 调用job文件
     * @param jobName
     * @throws Exception
     */
    public static void callNativeJob(String jobName) throws Exception{
        callNativeTransWithParams(null, jobName);
    }

    /**
     * 调用job文件
     * @param jobName
     * @throws Exception
     */
    public static void callNativeJobWithParams(Map<String,String> params,String jobName) throws Exception{
        // 初始化
        KettleEnvironment.init();
        JobMeta jobMeta = new JobMeta(jobName,null);
        Job job = new Job(null, jobMeta);
        //向Job 脚本传递参数，脚本中获取参数值：${参数名}
        //参数
        if(params != null){
            for(Map.Entry<String,String> entry:  params.entrySet() ) {
                job.setVariable(entry.getKey(), entry.getValue());
            }
        }
        job.start();
        job.waitUntilFinished();
        if (job.getErrors() > 0) {
            throw new RuntimeException("Kettle执行作业发生异常");
        }
    }
}
